//---------------------------------------------------------- -*- Mode: C++ -*-
// $Id: ClientSM.cc 385 2010-05-27 15:58:30Z sriramsrao $
//
// Created 2006/03/23
//
// Copyright 2008 Quantcast Corp.
// Copyright 2006-2008 Kosmix Corp.
//
// This file is part of Kosmos File System (KFS).
//
// Licensed under the Apache License, Version 2.0
// (the "License"); you may not use this file except in compliance with
// the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
//
//
//----------------------------------------------------------------------------

#include "ClientSM.h"

#include "ChunkManager.h"
#include "ChunkServer.h"
#include "Utils.h"
#include "KfsOps.h"
#include "AtomicRecordAppender.h"

#include <string>
#include <sstream>
using std::string;
using std::ostringstream;

#include "common/log.h"
#include "libkfsIO/Globals.h"
#include "DiskIo.h"
#include "qcdio/qcutils.h"

#define CLIENT_SM_LOG_STREAM_PREFIX << "I" << mInstanceNum << "I " << GetPeerName() << " "
#define CLIENT_SM_LOG_STREAM(pri)  KFS_LOG_STREAM(pri)  CLIENT_SM_LOG_STREAM_PREFIX
#define CLIENT_SM_LOG_STREAM_DEBUG KFS_LOG_STREAM_DEBUG CLIENT_SM_LOG_STREAM_PREFIX
#define CLIENT_SM_LOG_STREAM_WARN  KFS_LOG_STREAM_WARN  CLIENT_SM_LOG_STREAM_PREFIX
#define CLIENT_SM_LOG_STREAM_INFO  KFS_LOG_STREAM_INFO  CLIENT_SM_LOG_STREAM_PREFIX
#define CLIENT_SM_LOG_STREAM_ERROR KFS_LOG_STREAM_ERROR CLIENT_SM_LOG_STREAM_PREFIX
#define CLIENT_SM_LOG_STREAM_FATAL KFS_LOG_STREAM_FATAL CLIENT_SM_LOG_STREAM_PREFIX

using namespace KFS;
using namespace KFS::libkfsio;

const int kMaxCmdHeaderLength = 1 << 10;
bool ClientSM::sTraceRequestResponse = false;
uint64_t ClientSM::sInstanceNum = 10000;

inline std::string ClientSM::GetPeerName()
{
    return (mNetConnection ?
        mNetConnection->GetPeerName() :
        std::string("not connected")
    );
}

inline BufferManager& ClientSM::GetBufferManager()
{
    return DiskIo::GetBufferManager();
}

inline void ClientSM::SendResponse(KfsOp* op, ClientSM::ByteCount opBytes)
{
    ByteCount respBytes = 0;
    respBytes = mNetConnection->GetNumBytesToWrite();
    SendResponse(op);
    respBytes = std::max(ByteCount(0),
        mNetConnection->GetNumBytesToWrite() - respBytes);
    mPrevNumToWrite = mNetConnection->GetNumBytesToWrite();
    GetBufferManager().Put(*this, opBytes - respBytes);
}

ClientSM::ClientSM(NetConnectionPtr &conn)
    : mNetConnection(conn),
      mCurOp(0),
      mPrevNumToWrite(0),
      mRecursionCnt(0),
      mInstanceNum(sInstanceNum++)
{
    SET_HANDLER(this, &ClientSM::HandleRequest);
    mNetConnection->SetMaxReadAhead(kMaxCmdHeaderLength);
    mNetConnection->SetInactivityTimeout(gClientManager.GetIdleTimeoutSec());
}

ClientSM::~ClientSM()
{
    KfsOp *op;

    assert(mOps.empty() && mPendingOps.empty() && mPendingSubmitQueue.empty());
    while (!mOps.empty()) {
        op = mOps.front().first;
        mOps.pop_front();
        delete op;
    }
    while (!mPendingOps.empty()) {
        op = mPendingOps.front().dependentOp;
        mPendingOps.pop_front();
        delete op;
    }
    while (!mPendingSubmitQueue.empty()) {
        op = mPendingSubmitQueue.front().dependentOp;
        mPendingSubmitQueue.pop_front();
        delete op;
    }
    gClientManager.Remove(this);
    // CLIENT_SM_LOG_STREAM_DEBUG << "~ClientSM" << KFS_LOG_EOM;
}

///
/// Send out the response to the client request.  The response is
/// generated by MetaRequest as per the protocol.
/// @param[in] op The request for which we finished execution.
///
void
ClientSM::SendResponse(KfsOp *op)
{
    assert(mNetConnection && op);

    struct timeval now;
    gettimeofday(&now, 0);
    const double timespent = ComputeTimeDiff(op->startTime, now);
    const bool   tooLong   = timespent > 0.2;
    CLIENT_SM_LOG_STREAM(op->status >= 0 ?
            (tooLong ? MsgLogger::kLogLevelINFO : MsgLogger::kLogLevelDEBUG) :
            MsgLogger::kLogLevelERROR) <<
        "seq: "        << op->seq <<
        " status: "    << op->status <<
        " buffers: "   << GetByteCount() <<
        " " << op->Show() <<
        (op->statusMsg.empty() ? "" : " msg: ") << op->statusMsg <<
        (tooLong ? " RPC too long " : " took: ") << timespent << " sec." <<
    KFS_LOG_EOM;

    IOBuffer::OStream os;
    op->Response(os);
    mNetConnection->Write(&os);
    IOBuffer* iobuf = 0;
    int       len   = 0;
    op->ResponseContent(iobuf, len);
    mNetConnection->Write(iobuf, len);
    gClientManager.RequestDone((int64_t)(timespent * 1e6), *op);
}

///
/// Generic event handler.  Decode the event that occurred and
/// appropriately extract out the data and deal with the event.
/// @param[in] code: The type of event that occurred
/// @param[in] data: Data being passed in relative to the event that
/// occurred.
/// @retval 0 to indicate successful event handling; -1 otherwise.
///
int
ClientSM::HandleRequest(int code, void *data)
{
    IOBuffer *iobuf;
    KfsOp *op;
    int cmdLen = 0;
    bool gotCmd = false;

#ifdef DEBUG
    verifyExecutingOnNetProcessor();
#endif
    assert(mRecursionCnt >= 0 && mNetConnection);
    mRecursionCnt++;

    switch (code) {
    case EVENT_NET_READ:
        if (IsWaiting()) {
            CLIENT_SM_LOG_STREAM_DEBUG <<
                "spurious read: " << (mCurOp ? mCurOp->Show() : "cmd") <<
                " waiting for: " << GetByteCount() <<
                " bytes of io buffers" <<
            KFS_LOG_EOM;
            mNetConnection->SetMaxReadAhead(0);
            break;
        }
	// We read something from the network.  Run the RPC that
	// came in.
	iobuf = (IOBuffer *) data;
	while ((mCurOp || IsMsgAvail(iobuf, &cmdLen)) &&
                (gotCmd = HandleClientCmd(iobuf, cmdLen))) {
            cmdLen = 0;
            gotCmd = 0;
	}
        if (! mCurOp) {
            int hdrsz;
            if (cmdLen > 0 && ! gotCmd) {
                CLIENT_SM_LOG_STREAM_ERROR <<
                    " failed to parse request, closing connection;"
                    " header size: "    << cmdLen <<
                    " read available: " << iobuf->BytesConsumable() <<
                KFS_LOG_EOM;
                gClientManager.BadRequest();
            } else if ((hdrsz = iobuf->BytesConsumable()) > MAX_RPC_HEADER_LEN) {
                CLIENT_SM_LOG_STREAM_ERROR <<
                    " exceeded max request header size: " << hdrsz <<
                    " limit: " << MAX_RPC_HEADER_LEN <<
                    ", closing connection" <<
                KFS_LOG_EOM;
                gClientManager.BadRequestHeader();
            } else {
                break;
            }
            iobuf->Clear();
            mNetConnection->Close();
        }
	break;

    case EVENT_NET_WROTE: {
            const int rem = mNetConnection->GetNumBytesToWrite();
            GetBufferManager().Put(*this, mPrevNumToWrite - rem);
            mPrevNumToWrite = rem;
        }
	break;

    case EVENT_CMD_DONE:
        // An op finished execution.  Send response back in FIFO
        gChunkServer.OpFinished();
        op = (KfsOp *) data;
        op->done = true;
        assert(!mOps.empty());
        if (sTraceRequestResponse) {
            IOBuffer::OStream os;
            op->Response(os);
            IOBuffer::IStream is(os);
            string line;
            while (getline(is, line)) {
                CLIENT_SM_LOG_STREAM_DEBUG <<
                    "response: " << line <<
                KFS_LOG_EOM;
            }
        }
        while (!mOps.empty()) {
            KfsOp *qop = mOps.front().first;
            if (!qop->done) {
                if (! op) {
                    break;
                }
                if (op->op == CMD_WRITE_ID_ALLOC ||
                        op->op == CMD_RECORD_APPEND) {
                    OpsQueue::iterator i;
                    for (i = mOps.begin(); i != mOps.end() && op != i->first; ++i)
                        {}
                    assert(i != mOps.end() && op == i->first);
                    assert(mPendingOps.empty() || op->seq != mPendingOps.front().op->seq);
                    if (i != mOps.end()) {
                        SendResponse(op, i->second);
                    }
                    if (i != mOps.end()) {
                        mOps.erase(i);
                        OpFinished(op);
                    }
                    delete op;
                } else {
                    CLIENT_SM_LOG_STREAM_DEBUG <<
                        "previous op still pending: " <<
                        qop->Show() << "; deferring reply to: " <<
                        op->Show() <<
                    KFS_LOG_EOM;
                }
                break;
            }
            if (qop == op) {
                op = 0;
            }
            SendResponse(qop, mOps.front().second);
            mOps.pop_front();
            OpFinished(qop);
            delete qop;
        }
        mNetConnection->StartFlush();
        break;

    case EVENT_INACTIVITY_TIMEOUT:
    case EVENT_NET_ERROR:
        CLIENT_SM_LOG_STREAM_DEBUG <<
            "closing connection"
            " due to " << (code == EVENT_INACTIVITY_TIMEOUT ?
                "inactivity timeout" : "network error") <<
            ", socket error: " <<
                QCUtils::SysError(mNetConnection->GetSocketError()) <<
            ", pending read: " << mNetConnection->GetNumBytesToRead() <<
            " write: " << mNetConnection->GetNumBytesToWrite() <<
        KFS_LOG_EOM;
	mNetConnection->Close();
	break;

    default:
	assert(!"Unknown event");
	break;
    }

    assert(mRecursionCnt > 0);
    if (mRecursionCnt == 1) {
        if (mNetConnection->IsGood()) {
            // Enforce 5 min timeout if connection has pending read and write.
            mNetConnection->SetInactivityTimeout(
                (mNetConnection->HasPendingRead() ||
                    mNetConnection->IsWriteReady()) ?
                gClientManager.GetIoTimeoutSec() :
                gClientManager.GetIdleTimeoutSec());
        } else {
            std::list<RemoteSyncSMPtr> serversToRelease;

            mRemoteSyncers.swap(serversToRelease);
            // get rid of the connection to all the peers in daisy chain;
            // if there were any outstanding ops, they will all come back
            // to this method as EVENT_CMD_DONE and we clean them up above.
            ReleaseAllServers(serversToRelease);
            ReleaseChunkSpaceReservations();
            mRecursionCnt--;
            // if there are any disk ops, wait for the ops to finish
            SET_HANDLER(this, &ClientSM::HandleTerminate);
            HandleTerminate(EVENT_NET_ERROR, NULL);
            // this can be deleted, return now.
            return 0;
        }
    }
    mRecursionCnt--;
    return 0;
}

///
/// Termination handler.  For the client state machine, we could have
/// ops queued at the logger.  So, for cleanup wait for all the
/// outstanding ops to finish and then delete this.  In this state,
/// the only event that gets raised is that an op finished; anything
/// else is bad.
///
int
ClientSM::HandleTerminate(int code, void *data)
{
    KfsOp *op;

#ifdef DEBUG
    verifyExecutingOnNetProcessor();
#endif

    switch (code) {
    case EVENT_CMD_DONE:
        gChunkServer.OpFinished();
	// An op finished execution.  Send a response back
	op = (KfsOp *) data;
	op->done = true;
	if (op != mOps.front().first)
	    break;
	while (!mOps.empty()) {
	    op = mOps.front().first;
	    if (!op->done)
		break;
            GetBufferManager().Put(*this, mOps.front().second);
            OpFinished(op);
	    // we are done with the op
	    mOps.pop_front();
	    delete op;
	}
	break;

    case EVENT_INACTIVITY_TIMEOUT:
    case EVENT_NET_ERROR:
        // clean things up
        break;

    default:
	assert(!"Unknown event");
	break;
    }

    if (mOps.empty()) {
        // all ops are done...so, now, we can nuke ourself.
        assert(mPendingOps.empty());
        if (mNetConnection) {
            mNetConnection->SetOwningKfsCallbackObj(0);
        }
        delete this;
        return 1;
    }
    return 0;
}

template <typename T> bool
ClientSM::GetWriteOp(T* wop, int align, IOBuffer *iobuf, int cmdLen, IOBuffer*& ioOpBuf)
{
    const size_t nAvail = iobuf->BytesConsumable() - cmdLen;
    if (nAvail < wop->numBytes) {
        if (! mCurOp) {
            BufferManager& bufMgr = GetBufferManager();
            bool overQuota = false;
            if (wop->numBytes > gChunkManager.GetMaxIORequestSize() ||
                    (overQuota = bufMgr.IsOverQuota(*this, wop->numBytes))) {
                CLIENT_SM_LOG_STREAM_ERROR <<
                    "seq: " << wop->seq <<
                    " bad write request size: " << wop->numBytes <<
                    " buffers: " << GetByteCount() <<
                    (overQuota ? " over quota" : "") <<
                    ", closing connection" <<
                KFS_LOG_EOM;
                delete wop;
                return false;
            }
            // Move write data to the start of the buffers, to make it
            // aligned. Normally only one buffer will be created.
            iobuf->Consume(cmdLen);
            const int off(align % IOBufferData::GetDefaultBufferSize());
            if (off > 0) {
                IOBuffer buf;
                buf.ReplaceKeepBuffersFull(iobuf, off, nAvail);
                iobuf->Move(&buf);
                iobuf->Consume(off);
            } else {
                iobuf->MakeBuffersFull();
            }
            mCurOp = wop;
            if (! bufMgr.Get(*this, wop->numBytes)) {
                CLIENT_SM_LOG_STREAM_DEBUG <<
                    "seq: " << wop->seq <<
                    " request for: " << wop->numBytes << " bytes denied" <<
                    " cur: "   << GetByteCount() <<
                    " total: " << bufMgr.GetTotalByteCount() <<
                    " used: "  << bufMgr.GetUsedByteCount() <<
                    " bufs: "  << bufMgr.GetFreeBufferCount() <<
                    " op: " << wop->Show() <<
                    " waiting for buffers" <<
                KFS_LOG_EOM;
                mNetConnection->SetMaxReadAhead(0);
                return false;
            }
        }
        mNetConnection->SetMaxReadAhead(wop->numBytes - nAvail);
        // we couldn't process the command...so, wait
        return false;
    }
    if (ioOpBuf) {
        ioOpBuf->Clear();
    } else {
        ioOpBuf = new IOBuffer();
    }
    if (nAvail != wop->numBytes || cmdLen > 0) {
        assert(nAvail >= wop->numBytes);
        iobuf->Consume(cmdLen);
        const int off(align % IOBufferData::GetDefaultBufferSize());
        if (nAvail == wop->numBytes && off <= 0) {
            iobuf->MakeBuffersFull();
            ioOpBuf->Move(iobuf);
        } else {
            ioOpBuf->ReplaceKeepBuffersFull(iobuf, off, wop->numBytes);
            ioOpBuf->Consume(off);
        }
    } else {
        ioOpBuf->Move(iobuf);
    }
    mCurOp = 0;
    mNetConnection->SetMaxReadAhead(kMaxCmdHeaderLength);
    return true;
}

inline BufferManager::ByteCount
IoRequestBytes(size_t numBytes)
{
    if (numBytes <= 0) {
        return 0;
    }
    return ((numBytes + KFS::CHECKSUM_BLOCKSIZE - 1) /
        KFS::CHECKSUM_BLOCKSIZE * KFS::CHECKSUM_BLOCKSIZE);
}

///
/// We have a command in a buffer.  It is possible that we don't have
/// everything we need to execute it (for example, for a write we may
/// not have received all the data the client promised).  So, parse
/// out the command and if we have everything execute it.
///
bool
ClientSM::HandleClientCmd(IOBuffer *iobuf, int cmdLen)
{
    KfsOp *op = mCurOp;

    assert(op ? cmdLen == 0 : cmdLen > 0);
    if (! op) {
        if (sTraceRequestResponse) {
            IOBuffer::IStream is(*iobuf, cmdLen);
            string line;
            while (getline(is, line)) {
                CLIENT_SM_LOG_STREAM_DEBUG <<
                    "request: " << line <<
                KFS_LOG_EOM;
            }
        }
        IOBuffer::IStream is(*iobuf, cmdLen);
        if (ParseCommand(is, &op) != 0) {
            assert(! op);
            is.Rewind(cmdLen);
            string line;
            int    maxLines = 64;
            while (--maxLines >= 0 && getline(is, line)) {
                CLIENT_SM_LOG_STREAM_ERROR <<
                    "invalid request: " << line <<
                KFS_LOG_EOM;
            }
            iobuf->Consume(cmdLen);
            // got a bogus command
            return false;
        }
    }

    ByteCount bufferBytes = -1;
    if (op->op == CMD_WRITE_PREPARE) {
        WritePrepareOp* const wop = static_cast<WritePrepareOp*>(op);
        assert(! wop->dataBuf);
        if (! GetWriteOp(wop, wop->offset, iobuf, cmdLen, wop->dataBuf)) {
            return false;
        }
        bufferBytes = IoRequestBytes(wop->numBytes);
    } else if (op->op == CMD_RECORD_APPEND) {
        RecordAppendOp* const waop = static_cast<RecordAppendOp*>(op);
        IOBuffer* opBuf = &waop->dataBuf;
        if (! GetWriteOp(
                waop,
                mCurOp ? 0 :
                    gAtomicRecordAppendManager.GetAlignment(waop->chunkId),
                iobuf,
                cmdLen,
                opBuf
            )) {
            return false;
        }
        assert(opBuf == &waop->dataBuf);
        bufferBytes = IoRequestBytes(waop->numBytes);
    } else {
        iobuf->Consume(cmdLen);
    }
    CLIENT_SM_LOG_STREAM_DEBUG <<
        "got: seq: " << op->seq << " " << op->Show() <<
    KFS_LOG_EOM;

    bool submitResponseFlag = false;
    if (op->op == CMD_READ) {
        ReadOp* const rop = static_cast<ReadOp*>(op);
        bufferBytes = IoRequestBytes(rop->numBytes);
        if (! mCurOp) {
            BufferManager& bufMgr = GetBufferManager();
            if (bufMgr.IsOverQuota(*this, bufferBytes)) {
                CLIENT_SM_LOG_STREAM_ERROR <<
                    " bad read request size: " << rop->numBytes <<
                    " need: " << bufferBytes <<
                    " buffers: " << GetByteCount() <<
                    " over quota, closing connection" <<
                KFS_LOG_EOM;
                delete rop;
                return false;
            }
            if (! bufMgr.Get(*this, bufferBytes)) {
                mCurOp = op;
                CLIENT_SM_LOG_STREAM_DEBUG <<
                    "request for: " << bufferBytes << " bytes denied" <<
                    " cur: "   << GetByteCount() <<
                    " total: " << bufMgr.GetTotalByteCount() <<
                    " used: "  << bufMgr.GetUsedByteCount() <<
                    " bufs: "  << bufMgr.GetFreeBufferCount() <<
                    " op: "    << op->Show() <<
                    " waiting for buffers" <<
                KFS_LOG_EOM;
                mNetConnection->SetMaxReadAhead(0);
                return false;
            }
            mNetConnection->SetMaxReadAhead(kMaxCmdHeaderLength);
        }
        mCurOp = 0;

        if (! gChunkManager.IsChunkStable(rop->chunkId)) {
            // Do not allow dirty reads.
            rop->statusMsg = "chunk not stable";
            rop->status    = -EAGAIN;
            submitResponseFlag = true;
            CLIENT_SM_LOG_STREAM_ERROR <<
                " read request for unstable chunk: " << rop->chunkId <<
                " denied" <<
            KFS_LOG_EOM;
        }
    }

    if (bufferBytes < 0) {
        assert(
            op->op != CMD_WRITE_PREPARE &&
            op->op != CMD_RECORD_APPEND &&
            op->op != CMD_READ
        );
        // This is needed to account for large number of small responses to
        // prevent out of buffers in the case where the client queues requests
        // but doesn't read replies.
        const int kLargeReponseHeaderSize = 4 << 10;
        bufferBytes = kLargeReponseHeaderSize;
        BufferManager& bufMgr = GetBufferManager();
        if (! bufMgr.Get(*this, bufferBytes)) {
            mCurOp = op;
            CLIENT_SM_LOG_STREAM_DEBUG <<
                "request for: " << bufferBytes << " bytes denied" <<
                " cur: "   << GetByteCount() <<
                " total: " << bufMgr.GetTotalByteCount() <<
                " used: "  << bufMgr.GetUsedByteCount() <<
                " bufs: "  << bufMgr.GetFreeBufferCount() <<
                " op: "    << op->Show() <<
                " waiting for buffers" <<
            KFS_LOG_EOM;
            mNetConnection->SetMaxReadAhead(0);
            return false;
        }
        mNetConnection->SetMaxReadAhead(kMaxCmdHeaderLength);
        mCurOp = 0;
    }

    if (op->op == CMD_WRITE_SYNC) {
        // make the write sync depend on a previous write
        KfsOp *w = NULL;
        for (OpsQueue::iterator i = mOps.begin(); i != mOps.end(); i++) {
            if (i->first->op == CMD_WRITE_PREPARE ||
                    i->first->op == CMD_WRITE_PREPARE_FWD ||
                    i->first->op == CMD_WRITE) {
                w = i->first;
            }
        }
        if (w != NULL) {
            OpPair p;

            op->clnt = this;
            p.op = w;
            p.dependentOp = op;
            mPendingOps.push_back(p);

            CLIENT_SM_LOG_STREAM_DEBUG <<
                "keeping write-sync (" << op->seq <<
                ") pending and depends on " << w->seq <<
            KFS_LOG_EOM;
            return true;
        } else {
            CLIENT_SM_LOG_STREAM_DEBUG <<
                "write-sync is being pushed down; no writes left, "
                << mOps.size() << " ops left" <<
            KFS_LOG_EOM;
        }
    }

    mOps.push_back(std::make_pair(op, bufferBytes));
    op->clnt = this;
    gChunkServer.OpInserted();
    if (submitResponseFlag) {
        HandleRequest(EVENT_CMD_DONE, op);
    } else {
        // op->Execute();
        SubmitOp(op);
    }
    return true;
}

void
ClientSM::OpFinished(KfsOp *doneOp)
{
    // Multiple ops could be waiting for a single op to finish.
    //
    // Do not run pending submit queue here, if it is not empty.
    // If pending submit is not empty here, then this is recursive call. Just
    // add the op to the pending submit queue and let the caller run the queue.
    // This is need to send responses in the request order, and to limit the
    // recursion depth.
    const bool runPendingSubmitQueueFlag = mPendingSubmitQueue.empty();
    while (! mPendingOps.empty()) {
        OpPair& p = mPendingOps.front();
        if (p.op != doneOp) {
            break;
        }
        CLIENT_SM_LOG_STREAM_DEBUG <<
            "submitting write-sync (" << p.dependentOp->seq <<
            ") since " << p.op->seq << " finished" <<
        KFS_LOG_EOM;
        mPendingSubmitQueue.splice(mPendingSubmitQueue.end(),
            mPendingOps, mPendingOps.begin());
    }
    if (! runPendingSubmitQueueFlag) {
        return;
    }
    while (! mPendingSubmitQueue.empty()) {
        KfsOp* const op = mPendingSubmitQueue.front().dependentOp;
        mPendingSubmitQueue.pop_front();
        gChunkServer.OpInserted();
        mOps.push_back(std::make_pair(op, 0));
        SubmitOp(op);
    }
}

void
ClientSM::ReleaseChunkSpaceReservations()
{
    for (ChunkSpaceResMap::iterator iter = mReservations.begin();
         iter != mReservations.end(); iter++) {
        gAtomicRecordAppendManager.ChunkSpaceRelease(
            iter->first.chunkId, iter->first.transactionId, iter->second);
    }
}

RemoteSyncSMPtr
ClientSM::FindServer(const ServerLocation &loc, bool connect)
{
    return KFS::FindServer(mRemoteSyncers, loc, connect);
}

void
ClientSM::Granted(ClientSM::ByteCount byteCount)
{
    CLIENT_SM_LOG_STREAM_DEBUG << "granted: " << byteCount << " op: " <<
        (mCurOp ? mCurOp->Show() : std::string("null")) <<
    KFS_LOG_EOM;
    if (! mNetConnection) {
        return;
    }
    if (mCurOp) {
        IOBuffer buf;
        ClientSM::HandleClientCmd(&buf, 0);
    } else {
        mNetConnection->SetMaxReadAhead(kMaxCmdHeaderLength);
    }
}
